package net.bwie.douyin.dwd.log.Function;

import net.bwie.douyin.dwd.log.bean.LiveEvent;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class robotSpeakFunction implements WindowFunction<LiveEvent, String, String, TimeWindow> {


    private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");
    @Override
    public void apply(String s, TimeWindow window, Iterable<LiveEvent> input, Collector<String> out) throws Exception {
        String windowStartTime = format.format(window.getStart());
        String windowEndTime = format.format(window.getEnd());

        int Count = 0;

        for (LiveEvent liveEvent : input) {
            Count++;
        }

        if (Count>=2){
            out.collect(windowStartTime + "," + windowEndTime + "," + s  + "," + Count+","+System.currentTimeMillis());
        }
    }
}
